package com.gitee.zycra.future.job.client.executor;

import com.gitee.zycra.future.job.client.api.JobHandler;
import com.gitee.zycra.future.job.client.container.SystemInfoContainer;
import com.gitee.zycra.future.job.client.dto.JobTriggerDTO;
import com.gitee.zycra.future.job.client.dto.NodeStatusDTO;
import com.gitee.zycra.future.job.client.dto.RunningStatusDTO;
import com.gitee.zycra.future.job.client.enums.ScheduleRatioEnum;
import com.gitee.zycra.future.job.client.util.JSONUtil;
import lombok.extern.slf4j.Slf4j;
import okhttp3.OkHttpClient;
import okhttp3.Request;
import okhttp3.Response;
import okhttp3.WebSocket;
import okhttp3.WebSocketListener;

import java.time.LocalDateTime;
import java.time.ZoneId;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * @author zycra
 */
@Slf4j
public final class SyncStatusExecutor implements Runnable {

    private static final Map<Long, Integer> RUNNING_MAP = new ConcurrentHashMap<>();

    private static final Map<Long, ScheduledFuture<?>> RUNNING_FUTURE_MAP = new ConcurrentHashMap<>();

    public static final int MAX_SCHEDULE_RATIO = 100;

    private final OkHttpClient client;

    private final Request request;

    private WebSocket webSocket;

    private final ScheduledThreadPoolExecutor jobExecutor = new ScheduledThreadPoolExecutor(10);

    private boolean connected;

    public boolean getConnected() {
        return connected;
    }

    public SyncStatusExecutor setConnected(boolean connected) {
        this.connected = connected;
        return this;
    }

    public SyncStatusExecutor(String host, String applicationName) {
        this.client = new OkHttpClient.Builder().build();
        this.request = new Request.Builder().get().url(host + "/syncJobInfo?applicationName=" + applicationName).build();
    }

    @Override
    public void run() {
        if (!getConnected()) {
            webSocket = client.newWebSocket(request, createListener());
        }
        if (!getConnected()) {
            log.info("重连失败, 等待下次重试");
            return;
        }
        int cpuUsedRatio = SystemInfoContainer.CPU_USED_RATIO.get();
        int memoryUsedRatio = SystemInfoContainer.MEMORY_USED_RATIO.get();
        List<RunningStatusDTO> runningList = new ArrayList<>();
        for (Map.Entry<Long, Integer> entry : RUNNING_MAP.entrySet()) {
            runningList.add(new RunningStatusDTO().setRunningId(entry.getKey()).setScheduleRatio(entry.getValue()));
        }
        boolean success = webSocket.send(JSONUtil.toJSONString(new NodeStatusDTO()
                .setRunningList(runningList)
                .setCpuUsedRatio(cpuUsedRatio)
                .setMemoryUsedRatio(memoryUsedRatio)));
        if (success) {
            for (RunningStatusDTO runningStatusDTO : runningList) {
                Integer scheduleRatio = runningStatusDTO.getScheduleRatio();
                if (ScheduleRatioEnum.EXECUTE_FAILED.getCode().equals(scheduleRatio)
                        || ScheduleRatioEnum.CANCELED.getCode().equals(scheduleRatio)
                        || ScheduleRatioEnum.TIMEOUT.getCode().equals(scheduleRatio)
                        || ScheduleRatioEnum.EXECUTE_SUCCESS.getCode().equals(scheduleRatio)) {
                    cancelRunning(runningStatusDTO.getRunningId());
                    RUNNING_MAP.remove(runningStatusDTO.getRunningId());
                }
            }
        }
    }


    private WebSocketListener createListener() {
        return new WebSocketListener() {
            @Override
            public void onOpen(WebSocket webSocket, Response response) {
                log.info("socket 连接成功");
                setConnected(true);
            }

            @Override
            public void onMessage(WebSocket webSocket, String text) {
                log.info("接收到调度通知, text={}", text);
                JobTriggerDTO jobTriggerDTO = JSONUtil.parseObject(text, JobTriggerDTO.class);
                Long runningId = jobTriggerDTO.getRunningId();
                if (Boolean.TRUE.equals(jobTriggerDTO.getCancel())) {
                    log.info("取消执行任务, jobTriggerDTO={}", jobTriggerDTO);
                    cancelRunning(runningId);
                    updateRunning(runningId, ScheduleRatioEnum.CANCELED.getCode());
                    return;
                }
                if (Boolean.TRUE.equals(jobTriggerDTO.getTimeout())) {
                    log.info("执行任务超时取消, jobTriggerDTO={}", jobTriggerDTO);
                    cancelRunning(runningId);
                    updateRunning(runningId, ScheduleRatioEnum.TIMEOUT.getCode());
                    return;
                }
                if (RUNNING_MAP.containsKey(runningId)) {
                    log.info("正在执行的任务, jobTriggerDTO={}", jobTriggerDTO);
                    return;
                }
                try {
                    Class<JobHandler> clazz = (Class<JobHandler>) Class.forName(jobTriggerDTO.getJobClassName());
                    JobHandler jobHandler = clazz.getDeclaredConstructor().newInstance();
                    jobHandler.setRunningId(runningId);
                    LocalDateTime scheduleTime = jobTriggerDTO.getScheduleTime();
                    if (scheduleTime == null) {
                        log.error("执行时间为空, jobTriggerDTO={}", jobTriggerDTO);
                        return;
                    }
                    ScheduledFuture<?> future = jobExecutor.schedule(() -> {
                        try {
                            jobHandler.run(jobTriggerDTO.getScheduleParam());
                            log.info("任务执行完成, 修改任务进度为成功, runningId={}", runningId);
                            updateRunning(runningId, MAX_SCHEDULE_RATIO);
                        } catch (Exception e) {
                            log.error("任务执行失败, jobTriggerDTO={}", jobTriggerDTO, e);
                            cancelRunning(runningId);
                            updateRunning(runningId, ScheduleRatioEnum.EXECUTE_FAILED.getCode());
                        }
                    }, getTimeDiff(jobTriggerDTO.getScheduleTime()), TimeUnit.MILLISECONDS);
                    RUNNING_MAP.put(runningId, 1);
                    RUNNING_FUTURE_MAP.put(runningId, future);
                } catch (Exception e) {
                    log.error("执行异常", e);
                    cancelRunning(runningId);
                    updateRunning(runningId, ScheduleRatioEnum.EXECUTE_FAILED.getCode());
                }
            }

            @Override
            public void onClosed(WebSocket webSocket, int code, String reason) {
                log.info("socket 连接已关闭");
            }

            @Override
            public void onFailure(WebSocket webSocket, Throwable t, Response response) {
                log.error("socket 连接失败, 尝试重连", t);
                setConnected(false);
            }
        };
    }

    private long getTimeDiff(LocalDateTime dateTime) {
        if (dateTime == null) {
            return 0L;
        }
        return Math.max(convertToTimestamp(dateTime) - convertToTimestamp(LocalDateTime.now()), 0L);
    }

    private long convertToTimestamp(LocalDateTime dateTime) {
        return dateTime.atZone(ZoneId.systemDefault()).toInstant().toEpochMilli();
    }

    public static void updateRunning(Long runningId, int scheduleRatio) {
        Integer currentRatio = RUNNING_MAP.get(runningId);
        if (currentRatio == null || (currentRatio > 0 && currentRatio < MAX_SCHEDULE_RATIO)) {
            RUNNING_MAP.put(runningId, Math.min(scheduleRatio, MAX_SCHEDULE_RATIO));
        }
    }

    private static void cancelRunning(Long runningId) {
        ScheduledFuture<?> future = RUNNING_FUTURE_MAP.get(runningId);
        if (future != null) {
            future.cancel(true);
            RUNNING_FUTURE_MAP.remove(runningId);
        }
    }
}
